package com.etl.webSocket;

import java.io.*;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.etl.entity.EtlTaskIncrement;
import com.etl.enums.ResultEnum;
import com.etl.mapper.EtlTaskIncrementMapper;
import com.etl.service.IEtlTaskIncrementService;
import com.etl.service.IEtlLogService;
import com.google.common.collect.Lists;
import com.etl.entity.EtlTaskMain;
import com.etl.mapper.EtlTaskMainMapper;
import com.etl.service.IEtlTaskMainService;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.Strings;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;


/**
 * @author zhengkai.blog.csdn.net
 */
@ServerEndpoint("/imserver")
@Component
@Slf4j
public class WebSocketServer {

    private IEtlLogService etlLogService;

    //此处是解决无法注入的关键
    private static ApplicationContext applicationContext;
    //要注入的service或者dao
    private IEtlTaskMainService etlMainTaskTaskService;

    //要注入的service或者dao
    private IEtlTaskIncrementService etlIncrementTaskTaskService;


    public static void setApplicationContext(ApplicationContext applicationContext) {
        WebSocketServer.applicationContext = applicationContext;
    }


    /**
     * 连接集合
     */
    private static Map<String, WebSocketUser> sessionMap = new ConcurrentHashMap<>();

    /**
     * 特殊信息匹配
     */
    //抽取任务开始
    public static final String TASK_BEGIN = "taskBegin";
    //抽取任务结束
    public static final String TASK_END = "taskEnd";
    //心跳检查
    public static final String HEART_MSG = "ping";
    //脚本执行开始
    public static final String MOUNT_TASK_BEGIN = "mountBegin";
    //在线人员
    public static final String ONLINE_USERS = "onlineUsers";
    //IM信息
    public static final String IM = "IM";

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session) {
        //添加到集合中
        WebSocketUser webSocketUser = new WebSocketUser();
        webSocketUser.setSession(session);
        sessionMap.put(session.getId(), webSocketUser);
        log.info("成功连接到客户端：{},当前 在线人数：{}", session.getId(), sessionMap.size());
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session) {
        //从集合中删除
        sessionMap.remove(session.getId());
        //获取在线人
        Iterator<String> i = sessionMap.keySet().iterator();
        List<JSONObject> list = Lists.newArrayList();
        while (i.hasNext()) {
            String id = i.next();
            WebSocketUser user = sessionMap.get(id);
            JSONObject obj = new JSONObject();
            obj.put("userName", user.getUserName());
            obj.put("id", id);
            obj.put("type", "loginOut");
            list.add(obj);
        }
        log.info("发送消息：{}", JSONObject.toJSONString(list));
        this.sendAll(JSONObject.toJSONString(list));
        log.info("连接断开");
    }


    /**
     * 发生错误时调用
     */
    @OnError
    public void onError(Session session, Throwable error) {
        error.printStackTrace();
    }

    /**
     * 封装一个send方法，发送消息到前端
     */
    private void send(Session session, String message) {
        try {
            session.getBasicRemote().sendText(message);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 封装一个sendAll方法，群发消息到前端
     */
    private void sendAll(String message) {
        try {
            for (String key : sessionMap.keySet()) {
                this.send(sessionMap.get(key).getSession(), message);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) throws IOException {
        log.info("来自客户端的消息：{}", message);
        JSONObject jsonObject = JSONObject.parseObject(message);
        String value = jsonObject.getString("type");
        WebSocketUser webSocketUser = sessionMap.get(session.getId());
        if (Strings.isNotBlank(value)) {
            switch (value) {
                case TASK_BEGIN: {
                    String interfaceType = jsonObject.getString("interfaceType");

                    if("increment".equals(interfaceType)){
                        Long kid = jsonObject.getLong("kid");
                        /**
                         * 防止一个窗口重复触发同一个任务
                         */
                        if (webSocketUser.getIsTaskRunning()) {
                            log.info("任务已经触发执行");
                            session.getBasicRemote().sendText("请勿重复触发任务，任务正在拼命执行中……");
                            return;
                        }
                        /**
                         * 检验任务，防止多个窗口同时触发同一个人任务
                         */
                        EtlTaskIncrementMapper etlTaskIncrementMapper = applicationContext.getBean(EtlTaskIncrementMapper.class);
                        QueryWrapper<EtlTaskIncrement> queryWrapper = new QueryWrapper<>();
                        queryWrapper.eq("KID", kid);
                        EtlTaskIncrement etlTaskIncrement = etlTaskIncrementMapper.selectOne(queryWrapper);
                        if (etlTaskIncrement == null) {
                            log.error("【执行单个任务】-{}", ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST.getErrMsg());
                            session.getBasicRemote().sendText(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST.getErrMsg());
                            return;
                        }
                        if (etlTaskIncrement.getExecFlag() == 1) {
                            log.error("【执行单个任务】-{}", ResultEnum.TASK_IS_RUNNING.getErrMsg());
                            session.getBasicRemote().sendText(ResultEnum.TASK_IS_RUNNING.getErrMsg() + ",如果确认任务假死，请重置任务状态并重新执行");
                            return;
                        }
                        log.info("开始读取任务日志");
                        webSocketUser.setIsTaskRunning(true);
                        //默认从第一行开始
                        webSocketUser.setLogLength(0);
                        String userName = jsonObject.getString("userName");
                        String systemName = jsonObject.getString("systemName");
                        String tableName = jsonObject.getString("tableName");
                        if (Strings.isNotBlank(userName) && Strings.isNotBlank(systemName) && Strings.isNotBlank(tableName)) {
                            this.sendAll(String.format("%s正在执行%s的%s表抽取，请勿重复执行！", userName, systemName, tableName));
                        }
                        log.info("执行任务");
                        etlIncrementTaskTaskService = applicationContext.getBean(IEtlTaskIncrementService.class);
                        etlIncrementTaskTaskService.socketExecuteSingle(session, kid);
                        applicationContext.getBean(IEtlLogService.class).readLogWebSocket(webSocketUser);
                    }else  if("main".equals(interfaceType)){

                        Long kid = jsonObject.getLong("kid");
                        /**
                         * 防止一个窗口重复触发同一个任务
                         */
                        if (webSocketUser.getIsTaskRunning()) {
                            log.info("任务已经触发执行");
                            session.getBasicRemote().sendText("请勿重复触发任务，任务正在拼命执行中……");
                            return;
                        }
                        /**
                         * 检验任务，防止多个窗口同时触发同一个人任务
                         */
                        EtlTaskMainMapper etlTaskMainMapper = applicationContext.getBean(EtlTaskMainMapper.class);
                        QueryWrapper<EtlTaskMain> queryWrapper = new QueryWrapper<>();
                        queryWrapper.eq("KID", kid);
                        EtlTaskMain etlTaskMain = etlTaskMainMapper.selectOne(queryWrapper);
                        if (etlTaskMain == null) {
                            log.error("【执行单个任务】-{}", ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST.getErrMsg());
                            session.getBasicRemote().sendText(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST.getErrMsg());
                            return;
                        }
                        if (etlTaskMain.getExecFlag() == 1) {
                            log.error("【执行单个任务】-{}", ResultEnum.TASK_IS_RUNNING.getErrMsg());
                            session.getBasicRemote().sendText(ResultEnum.TASK_IS_RUNNING.getErrMsg() + ",如果确认任务假死，请重置任务状态并重新执行");
                            return;
                        }
                        log.info("开始读取任务日志");
                        webSocketUser.setIsTaskRunning(true);
                        //默认从第一行开始
                        webSocketUser.setLogLength(0);
                        String userName = jsonObject.getString("userName");
                        String systemName = jsonObject.getString("systemName");
                        String tableName = jsonObject.getString("tableName");
                        if (Strings.isNotBlank(userName) && Strings.isNotBlank(systemName) && Strings.isNotBlank(tableName)) {
                            this.sendAll(String.format("%s正在执行%s的%s表抽取，请勿重复执行！", userName, systemName, tableName));
                        }
                        log.info("执行任务");
                        etlMainTaskTaskService = applicationContext.getBean(IEtlTaskMainService.class);
                        etlMainTaskTaskService.socketExecuteSingle(session, kid);
                        applicationContext.getBean(IEtlLogService.class).readLogWebSocket(webSocketUser);
                    }


                    break;
                }
                case TASK_END: {
                    webSocketUser.setIsTaskRunning(false);
                    webSocketUser.setLogLength(0);
                    break;
                }
                case HEART_MSG: {
                    session.getBasicRemote().sendText("pong");
                    break;
                }

                case ONLINE_USERS: {
                    String userName = jsonObject.getString("userName");
                    if (Strings.isBlank(webSocketUser.getUserName()) && Strings.isNotBlank(userName)) {
                        webSocketUser.setUserName(userName);
                    }
                    //获取在线人
                    Iterator<String> i = sessionMap.keySet().iterator();
                    List<JSONObject> list = Lists.newArrayList();
                    while (i.hasNext()) {
                        String id = i.next();
                        WebSocketUser user = sessionMap.get(id);
                        JSONObject obj = new JSONObject();
                        obj.put("userName", user.getUserName());
                        obj.put("id", id);
                        obj.put("type", "login");
                        list.add(obj);
                    }
                    log.info("发送消息：{}", JSONObject.toJSONString(list));
                    this.sendAll(JSONObject.toJSONString(list));
                    break;
                }
                case IM: {
                    String messages = jsonObject.getString("message");
                    JSONObject toUser = jsonObject.getJSONObject("toUser");
                    WebSocketUser toWebSocketUser = sessionMap.get(toUser.get("id"));
                    final String msgPrefix = "msg:";
                    //new StringBuilder(msgPrefix).append(webSocketUser.getUserName()).append(":").append(messages).toString()
                    toWebSocketUser.getSession().getBasicRemote().sendText(new StringBuilder(msgPrefix)
                            .append("<span style='color:red' >管理员</span>").append(":").append(messages).toString());
                    break;
                }
                default: {
                    log.error("没有匹配到socket消息规则");
                    break;
                }
            }
        }
    }







}
